package com.example.websocket.redis.config;

import com.example.websocket.redis.util.ClassUtils;
import com.example.websocket.redis.util.DataUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.stream.StreamReceiver;
import org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.time.Duration;

public abstract class AbstractRedisStreamReceiver<T> {

	private static final Logger logger = LogManager.getLogger(AbstractRedisStreamReceiver.class);
    @Autowired
    protected ReactiveRedisConnectionFactory connectionFactory;
    @Autowired
    protected ReactiveStringRedisTemplate redisTemplate;
    /**
     * 消费组名
     */
    protected String groupName;
    /**
     * 流名
     */
    protected String streamName;
    /**
     * 消费者名称
     */
    protected String name;
    
    protected Class<T> tClass;
    
    protected Consumer consumer;
    
    @SuppressWarnings("unchecked")
	protected AbstractRedisStreamReceiver(String streamName, String groupName, String name){
    	this.groupName = groupName;
    	this.streamName = streamName;
    	this.name = name;
    	this.consumer = Consumer.from(groupName, name);
        Class<?>[] cs = ClassUtils.getArgumentTypes(this.getClass());
        if (DataUtils.nonEmpty(cs)) {
            tClass = (Class<T>) cs[0];
        }
    }
    
    @PostConstruct
    protected void init() {
    	logger.debug("===============stream init=====================");
    	redisTemplate.hasKey(streamName).subscribe(isStreamExist -> {
    		if (isStreamExist) {
    			redisTemplate.opsForStream().groups(streamName)
    			.any(info -> info.groupName().equals(groupName))
    			.subscribe(isGroupExist -> {
    				if (!Boolean.TRUE.equals(isGroupExist)) {
    					logger.info("================流{}中，组{}不存在，自动创建====================", streamName, groupName);
    					redisTemplate.opsForStream().createGroup(streamName, groupName).subscribe();
    				}
    			});
    		} else {
				logger.info("================自动创建，流{}，组{}====================", streamName, groupName);
				redisTemplate.opsForStream().createGroup(streamName, groupName).subscribe();
    		}
    	});
        ackPEL();
        StreamReceiverOptions<String, ObjectRecord<String, T>> options = StreamReceiverOptions.builder()
            .pollTimeout(Duration.ofSeconds(1))
            .batchSize(200)
            .targetType(tClass)
            .build();
        StreamReceiver<String, ObjectRecord<String, T>> receiver = StreamReceiver.create(connectionFactory, options);
        StreamOffset<String> offset = StreamOffset.create(streamName, ReadOffset.lastConsumed());
        createReceive(receiver, offset).subscribe();      
    }
    /**
     * 创建具有错误还原的Flux
     */
    protected Flux<Long> createReceive(StreamReceiver<String, ObjectRecord<String, T>> receiver, StreamOffset<String> offset) {
        return receiver.receive(consumer, offset)
            .flatMap(this::onMessage)
            .onErrorResume(e -> {
            	logger.error("---------------流{}组{} StreamReceiver onErrorResume------------", streamName, groupName, e);
            	return createReceive(receiver, offset);
            });
    }
    /**
     * 处理读取的流数据，注意处理完数据后，调用：
     * redisTemplate.opsForStream().acknowledge(groupName, message)
     * 确认流数据已处理完成
     * @param message
     * @return
     */
    protected abstract Mono<Long> onMessage(ObjectRecord<String, T> message);
    /**
     * 标记未确认完成的数据列表的数据为已确认
     */
    @SuppressWarnings("unchecked")
    protected void ackPEL() {
        StreamOffset<String> pmlOffset = StreamOffset.fromStart(streamName);
        redisTemplate.opsForStream().read(tClass, consumer, pmlOffset)
            .flatMap(r -> {
            	logger.info("Ack record[{}] in the pending list", r.getId());
                return redisTemplate.opsForStream().acknowledge(groupName, r);
            }).subscribe();
    }
}
